package com.we.flink.deserialization;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** @Description: 通用的反序列化类 把数据转成 POJO 类型 目前只支持 JSON 格式的数据 */
public class PoJoDeserializationSchema<T> implements KafkaDeserializationSchema<T> {

    private static final Logger LOG = LoggerFactory.getLogger(PoJoDeserializationSchema.class);
    private final String encoding = "UTF8";
    private final Class<T> tClass;
    private final boolean includeMetadata;
    private final TypeInformation<T> type;

    public PoJoDeserializationSchema(Class<T> tClass, boolean includeMetadata) {
        this.tClass = tClass;
        this.includeMetadata = includeMetadata;
        this.type = TypeInformation.of(this.tClass);
    }

    @Override
    public boolean isEndOfStream(T jason) {
        return false;
    }

    @Override
    public T deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
        if (consumerRecord != null) {
            String value = new String(consumerRecord.value(), encoding);
            try {
                JSONObject jsonObject = JSONObject.parseObject(value);
                if (includeMetadata) {
                    String topic = consumerRecord.topic();
                    long timestamp = consumerRecord.timestamp();
                    jsonObject.put("topic", topic);
                    jsonObject.put("timestamp", timestamp);
                }
                return JSON.parseObject(String.valueOf(jsonObject), this.tClass);
            } catch (Exception e) {
                LOG.error(
                        "failed to parse json, original data: {}, error message: {}",
                        value,
                        e.getMessage());
            }
        }
        return null;
    }

    @Override
    public TypeInformation<T> getProducedType() {
        return this.type;
    }
}
